package com.favccxx.message.domain.message;

import com.alibaba.fastjson.JSON;
import com.favccxx.message.domain.message.adapter.CamelAdapter;
import com.favccxx.message.domain.message.entity.endpoint.MessageEndPoint;
import com.favccxx.message.domain.message.entity.endpoint.MessageEndPointRepo;
import com.favccxx.message.domain.message.entity.endpoint.RabbitmqEndPoint;
import com.favccxx.message.domain.message.entity.router.MessageRouter;
import com.favccxx.message.domain.message.entity.router.MessageRouterRepo;
import com.favccxx.message.domain.message.preprocesessor.MessagePreProcessor;
import com.favccxx.message.domain.message.processor.MessageProcessor;
import com.favccxx.message.infrastructure.util.jackson.JacksonUtil;
import com.favccxx.message.infrastructure.util.spring.SpringBeanUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.camel.*;
import org.apache.camel.builder.RouteBuilder;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.function.Function;
import java.util.stream.Collectors;

@Slf4j
@Service
public class MessageService {

    @Autowired
    MessageRouterRepo messageRouterRepo;
    @Autowired
    MessageEndPointRepo messageEndPointRepo;
    private Map<String, CamelAdapter> camelAdapterUnits;

    public MessageService(List<CamelAdapter> camelAdapters) {
        this.camelAdapterUnits = camelAdapters.stream().collect(Collectors.toMap(CamelAdapter::getDriverType, Function.identity()));
    }


    public List<CamelRouter> getCamelRouters(CamelContext camelContext) {
        List<MessageRouter> messageRouters = messageRouterRepo.findByStatus(true);

        List<CamelRouter> camelRouterList = messageRouters.stream().map(messageRouter -> {
            CamelRouter camelRouter = new CamelRouter();

            if (StringUtils.isBlank(messageRouter.getFromEndPoint()) || StringUtils.isBlank(messageRouter.getToEndPoint())) {
                log.info("##### Skip invalid message router -> " + messageRouter.getRouterName());
                return null;
            }

            Optional<MessageEndPoint> optFrom = messageEndPointRepo.findById(messageRouter.getFromEndPoint());
            if (optFrom.isEmpty()) {
                log.info("##### Skip message router, invalid message source -> " + messageRouter.getRouterName() + ":" + messageRouter.getFromEndPoint());
                return null;
            }
            Optional<MessageEndPoint> optTo = messageEndPointRepo.findById(messageRouter.getToEndPoint());
            if (optTo.isEmpty()) {
                log.info("##### Skip message router, invalid message destination -> " + messageRouter.getRouterName() + ":" + messageRouter.getToEndPoint());
                return null;
            }

            MessageEndPoint fromEndPoint = optFrom.get();
            if (fromEndPoint instanceof RabbitmqEndPoint) {
                log.info("true");
            }
            MessageEndPoint toEndPoint = optTo.get();

            String fromUri = createUri(camelContext, fromEndPoint);
            String toUri = createUri(camelContext, toEndPoint);

            camelRouter.setFrom(fromUri);
            camelRouter.setTo(toUri);


            Processor processor = new AsyncProcessor() {
                @Override
                public void process(Exchange exchange) throws Exception {
                    log.info("process void");
                    executeMessagePipeLine(exchange, camelRouter, messageRouter);
                }

                @Override
                public boolean process(Exchange exchange, AsyncCallback callback) {
                    log.info("process boolean");
                    executeMessagePipeLine(exchange, camelRouter, messageRouter);
                    callback.done(false);
                    return false;
                }

            /*    @Override
                public CompletableFuture<Exchange> processAsync(Exchange exchange) {
                    log.info("process CompletableFuture");
                    executeMessagePipeLine(exchange, camelRouter, messageRouter);
                    return null;
                }*/
            };
            camelRouter.setProcessor(processor);

            //    RouteBuilder routeBuilder = createRouteBuilder(camelRouter, processor);

       /*     camelRouter.setFilter(messageRouter.getFilterInstance());
            camelRouter.setPreprocessor(messageRouter.getPreprocessor());
            camelRouter.setPersistence(messageRouter.isPersistence());
            camelRouter.setProcessor(messageRouter.getProcessor());*//*     camelRouter.setFilter(messageRouter.getFilterInstance());
            camelRouter.setPreprocessor(messageRouter.getPreprocessor());
            camelRouter.setPersistence(messageRouter.isPersistence());
            camelRouter.setProcessor(messageRouter.getProcessor());*/

            return camelRouter;
        }).filter(x -> Objects.nonNull(x)).collect(Collectors.toList());

        return camelRouterList;
    }


    public void executeMessagePipeLine(Exchange exchange, CamelRouter camelRouter, MessageRouter messageRouter) {
        String preProcessorName = messageRouter.getPreprocessor();
        String data = (String) exchange.getIn().getBody();
        Object jsonResult = "";

        if (log.isDebugEnabled()) {
            log.debug("Message input data: " + data);
        }

        String format = messageRouter.getMessageFormat();
        if (MessageFormat.JSON.toString().equals(messageRouter.getMessageFormat())) {
            log.info("Ready deal json message");
        } else if (MessageFormat.XML.toString().equals(messageRouter.getMessageFormat())) {
            log.info("Ready deal XML message");
        }


        if (StringUtils.isNoneBlank(preProcessorName)) {
            MessagePreProcessor preProcessor = (MessagePreProcessor) SpringBeanUtil.getBean(preProcessorName);
            //preProcessor.process()
        }

        boolean localStorage = messageRouter.isPersistence();
        if (localStorage) {
            log.info("### save message to TSDB.");
        }

        String processorName = messageRouter.getProcessor();
        if (StringUtils.isNoneBlank(processorName)) {
            MessageProcessor processor = (MessageProcessor) SpringBeanUtil.getBean(processorName);
            jsonResult = processor.process(data, messageRouter.getProcessRules());
        }

        exchange.getIn().setBody(JacksonUtil.toJsonString(jsonResult));

    }


    /**
     * 创建Camel 消息的连接URI
     *
     * @param endPoint
     * @return
     */
    private String createUri(CamelContext camelContext, MessageEndPoint endPoint) {
        CamelAdapter camelAdapter = camelAdapterUnits.get(endPoint.getType());
        String uri = camelAdapter.getUri(camelContext, endPoint);
        return uri;
    }
}
